package com.gaxy.subject.infra.basic.es;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.ScoreSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;

/**
 * ES工具类
 *
 * @author Gaxy
 * @create 2024/06/27/下午 04:09
 */
@Component
@Slf4j
public class ESRestClient {

    /**
     * Key(String) -> 集群名  Value(RestHighLevelClient) -> 高级客户端库中的核心类
     * RestHighLevelClient包含了集群的ip地址和端口号，简化了与Elasticsearch的交互过程 无需直接构造HTTP请求
     */
    public static Map<String, RestHighLevelClient> clientMap = new HashMap<>();

    @Resource
    private ESConfigProperties esConfigProperties;

    /**
     * RequestOptions用来封装HTTP请求的各种选项，比如超时时间、重试策略、请求头信息等。
     */
    private static final RequestOptions COMMON_OPTIONS;

    /**
     * 初始化COMMON_OPTION
     * 静态代码块，在类加载时只会执行一次
     */
    static {
        // .DEFAULT表示获取系统提供的默认配置 .toBuilder()允许在这个默认配置的基础上进行修改或扩展，而不会直接修改默认配置本身
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
        // 构建RequestOptions实例COMMON_OPTIONS
        COMMON_OPTIONS = builder.build();
    }

    /**
     * 初始化ESRestClient
     * 初始化ESRestClient的clientMap，将集群名和集群的RestHighLevelClient对象对应
     */
    @PostConstruct  //@PostConstruct用于标记一个方法在实例初始化后执行，只能用于实例方法且方法必须是无参的
    public void initialize() {
        List<ESClusterConfig> esConfigs = esConfigProperties.getEsConfigs();
        for (ESClusterConfig esConfig : esConfigs) {
            log.info("initialize.config.name:{}, node:{}", esConfig.getName(), esConfig.getNodes());
            RestHighLevelClient restHighLevelClient = initRestClient(esConfig);
            if (restHighLevelClient != null) {
                clientMap.put(esConfig.getName(), restHighLevelClient);
            } else {
                log.error("initialize.config.name:{}, node:{}.initError", esConfig.getName(), esConfig.getNodes());
            }
        }
    }

    /**
     * 初始化RestHighLevelClient
     * 把ESClusterConfig中配置的ip地址+端口号 传到RestHighLevelClient对象中
     *
     * @param esClusterConfig
     * @return
     */
    private RestHighLevelClient initRestClient(ESClusterConfig esClusterConfig) {
        // 所有集群的ip地址+端口号数组 形式 ip:port -> 192.168.115.129:9200
        String[] ipPortArr = esClusterConfig.getNodes().split(",");

        // HttpHost代表一个HTTP主机信息，包括 scheme(如http或https)、host(主机名或IP地址) 和 port(端口号)。
        // 通常用于指定请求的目标服务器信息
        List<HttpHost> httpHostList = new ArrayList<>(ipPortArr.length);

        // 将每个集群的hostName和port存入httpHost对象
        for (String ipPort : ipPortArr) {
            //拆分ip和端口号
            String[] ipPortInfo = ipPort.split(":");
            if (ipPortInfo.length == 2) {
                HttpHost httpHost = new HttpHost(ipPortInfo[0], NumberUtils.toInt(ipPortInfo[1]));
                httpHostList.add(httpHost);
            }
        }
        // 将httpHostList列表复制到httpHosts数组
        HttpHost[] httpHosts = new HttpHost[httpHostList.size()];
        httpHostList.toArray(httpHosts);

        // 使用ES集群中所有节点的HTTP主机地址的数组创建RestHighLevelClient对象
        RestClientBuilder builder = RestClient.builder(httpHosts);
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(builder);
        return restHighLevelClient;
    }

    /**
     * 获取集群对应的RestHighLevelClient
     *
     * @return
     */
    private static RestHighLevelClient getClient(String clusterName) {
        return clientMap.get(clusterName);
    }

    /**
     * 新增文档
     *
     * @param esIndexInfo
     * @param esSourceData
     * @return
     */
    public static boolean insertDoc(ESIndexInfo esIndexInfo, ESSourceData esSourceData) {
        try {
            // IndexRequest类用于构建向ES索引中添加或更新文档的“请求”。
            // indexRequest对象封装了索引名称、文档数据、文档id
            IndexRequest indexRequest = new IndexRequest(esIndexInfo.getIndexName());
            indexRequest.source(esSourceData.getData());
            indexRequest.id(esSourceData.getDocId());
            // COMMON_OPTIONS是定义的RequestOptions实例，用来封装HTTP请求的各种选项，比如超时时间、重试策略、请求头信息等。
            // getClient(esIndexInfo.getClusterName())得到了该集群名对应的RestHighLevelClient，用于与ES交互
            // .index(indexRequest, COMMON_OPTIONS)执行实际的索引操作,将indexRequest中的文档数据发送到ES集群，根据COMMON_OPTIONS中设定的选项来处理该请求。
            getClient(esIndexInfo.getClusterName()).index(indexRequest, COMMON_OPTIONS);
            return true;
        } catch (Exception e) {
            log.error("insertDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 更新文档
     * @param esIndexInfo
     * @param esSourceData
     * @return
     */
    public static boolean updateDoc(ESIndexInfo esIndexInfo, ESSourceData esSourceData) {
        try {
            UpdateRequest updateRequest = new UpdateRequest();
            updateRequest.index(esIndexInfo.getIndexName());
            updateRequest.id(esSourceData.getDocId());
            updateRequest.doc(esSourceData.getData());
            getClient(esIndexInfo.getClusterName()).update(updateRequest, COMMON_OPTIONS);
            return true;
        } catch (Exception e) {
            log.error("updateDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 批量更新
     * @param esIndexInfo
     * @param esSourceDataList
     * @return
     */
    public static boolean batchUpdateDoc(ESIndexInfo esIndexInfo, List<ESSourceData> esSourceDataList) {
        try {
            boolean flag = false;

            // BulkRequest类用于构建批量操作请求
            // 将多个文档操作(如创建、更新或删除)组合到一个请求中，以减少网络往返时间和提高性能
            BulkRequest bulkRequest = new BulkRequest();

            for (ESSourceData esSourceData : esSourceDataList) {
                String docId = esSourceData.getDocId();
                if (StringUtils.isNotBlank(docId)){
                    UpdateRequest updateRequest = new UpdateRequest();
                    updateRequest.index(esIndexInfo.getIndexName());
                    updateRequest.id(esSourceData.getDocId());
                    updateRequest.doc(esSourceData.getData());
                    bulkRequest.add(updateRequest);
                    flag = true;
                }
            }
            if (flag){
                BulkResponse bulk = getClient(esIndexInfo.getClusterName()).bulk(bulkRequest, COMMON_OPTIONS);
                if (bulk.hasFailures()){
                    return false;
                }
            }
            return true;
        } catch (Exception e) {
            log.error("batchUpdateDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 删除所有文档
     *
     * @param esIndexInfo
     * @return
     */
    public static boolean delete(ESIndexInfo esIndexInfo) {
        try {
            // DeleteByQueryRequest类用于构建批量删除文档的“请求”，基于查询条件从索引中删除匹配的文档。
            DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(esIndexInfo.getIndexName());
            // deleteByQueryRequest.setQuery()设置DeleteByQueryRequest的查询条件
            // QueryBuilders.matchAllQuery()构建匹配所有文档的查询
            deleteByQueryRequest.setQuery(QueryBuilders.matchAllQuery());
            // BulkByScrollResponse对象封装了删除操作的响应
            BulkByScrollResponse response = getClient(esIndexInfo.getClusterName()).deleteByQuery(deleteByQueryRequest, COMMON_OPTIONS);
            // getDeleted()返回被成功删除的文档数量
            long deleted = response.getDeleted();
            log.info("delete.size:{}", JSON.toJSON(deleted));
            return true;
        } catch (Exception e) {
            log.error("delete.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 删除指定文档
     *
     * @param esIndexInfo
     * @param docId
     * @return
     */
    public static boolean deleteDoc(ESIndexInfo esIndexInfo, String docId) {
        try {
            // DeleteRequest类构用于建向ES发送删除“请求”
            DeleteRequest deleteRequest = new DeleteRequest(esIndexInfo.getIndexName());
            deleteRequest.id(docId);
            // DeleteResponse对象封装了删除操作的各种信息
            DeleteResponse response = getClient(esIndexInfo.getClusterName()).delete(deleteRequest, COMMON_OPTIONS);
            log.info("deleteDoc.response:{}", JSON.toJSONString(response));
            return true;
        } catch (Exception e) {
            log.error("deleteDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 查询指定文档是否存在
     *
     * @param esIndexInfo
     * @param docId
     * @return
     */
    public static boolean isExistDocById(ESIndexInfo esIndexInfo, String docId) {
        try {
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);
            return getClient(esIndexInfo.getClusterName()).exists(getRequest, COMMON_OPTIONS);
        } catch (Exception e) {
            log.error("isExistDocById.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 根据文档id获取源数据
     * @param esIndexInfo
     * @param docId
     * @return
     */
    public static Map<String, Object> getDocById(ESIndexInfo esIndexInfo, String docId) {
        try {
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);
            GetResponse response = getClient(esIndexInfo.getClusterName()).get(getRequest, COMMON_OPTIONS);
            Map<String, Object> source = response.getSource();
            return source;
        } catch (Exception e) {
            log.error("getDocById.exception:{}", e.getMessage(), e);
        }
        return null;
    }

    /**
     * 根据文档id和查询字段获取源数据
     * @param esIndexInfo
     * @param docId
     * @param fields
     * @return
     */
    public static Map<String, Object> getDocById(ESIndexInfo esIndexInfo, String docId, String[] fields) {
        try {
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);
            // 创建了FetchSourceContext对象，用于在ES中指定是否获取源文档以及获取哪些字段的源文档
            // true表示需要获取源文档, fields是字段列表, null表示不需要过滤源文档中的特定字段
            FetchSourceContext fetchSourceContext = new FetchSourceContext(true, fields, null);
            getRequest.fetchSourceContext(fetchSourceContext);
            GetResponse response = getClient(esIndexInfo.getClusterName()).get(getRequest, COMMON_OPTIONS);
            Map<String, Object> source = response.getSource();
            return source;
        } catch (Exception e) {
            log.error("getDocById.exception:{}", e.getMessage(), e);
        }
        return null;
    }

    /**
     * 条件搜索
     *
     * @param esIndexInfo
     * @param esSearchRequest 根据esSearchRequest对象中的查询条件和参数，构建搜索请求并执行搜索。
     * @return  搜索结果将作为SearchResponse对象返回
     */
    public static SearchResponse searchWithTermQuery(ESIndexInfo esIndexInfo, ESSearchRequest esSearchRequest) {
        try {
            // 从esSearchRequest对象中获取查询条件和参数
            BoolQueryBuilder bq = esSearchRequest.getBq();
            String[] fields = esSearchRequest.getFields();
            int from = esSearchRequest.getFrom();
            int size = esSearchRequest.getSize();
            Long minutes = esSearchRequest.getMinutes();
            Boolean needScroll = esSearchRequest.getNeedScroll();
            String sortName = esSearchRequest.getSortName();
            esSearchRequest.getSortOrder();

            // searchSourceBuilder构建搜索源，设置获取的查询条件和参数。
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(bq);
            searchSourceBuilder.fetchSource(fields, null).from(from).size(size);
            if (Objects.nonNull(esSearchRequest.getHighlightBuilder())){
                searchSourceBuilder.highlighter(esSearchRequest.getHighlightBuilder());
            }
            if (StringUtils.isNotBlank(sortName)){
                searchSourceBuilder.sort(sortName);
            }
            searchSourceBuilder.sort(new ScoreSortBuilder().order(SortOrder.DESC));

            // 创建SearchRequest对象，并设置搜索类型、索引名称，搜索源和滚动搜索
            SearchRequest searchRequest = new SearchRequest();
            searchRequest.searchType(SearchType.DEFAULT);
            searchRequest.indices(esIndexInfo.getIndexName());
            searchRequest.source(searchSourceBuilder);
            if (needScroll){
                Scroll scroll = new Scroll(TimeValue.timeValueMinutes(minutes));
                searchRequest.scroll(scroll);
            }

            // 得到了该集群名对应的RestHighLevelClient，执行搜索请求
            return getClient(esIndexInfo.getClusterName()).search(searchRequest, COMMON_OPTIONS);
        } catch (Exception e) {
            log.error("getDocById.exception:{}", e.getMessage(), e);
        }
        return null;
    }
}
